package com.project.base.common;

import com.project.base.core.ICallback;
import com.project.base.core.zookeeper.ZNodeEvent;
import org.apache.log4j.Logger;
import org.apache.zookeeper.*;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

/**
 * Zookeeper操作工具类，主要用于管理Zookeeper节点
 *
 * @author wxliyong
 */
public class ZookeeperUtils {
    private static final Logger LOGGER = Logger.getLogger(ZookeeperUtils.class);
    private static String zookeeperUrl = Constants.getInstance().getValue("zookeeper.url");
    //	private static String zookeeperUrl = "10.112.5.110:2181,10.112.5.220:2181,10.112.5.222:2181";
    private ZooKeeper zk;
    private static final int SESSION_TIMEOUT = 10000;
    private static ZookeeperUtils instance;

    private ZookeeperUtils() {
        init();
    }

    private void init() {
        try {
            CountDownLatch connectedLatch = new CountDownLatch(1);
            Watcher watcher = new ConnectedWatcher(connectedLatch);
            zk = new ZooKeeper(zookeeperUrl, SESSION_TIMEOUT, watcher);
            waitUntilConnected(zk, connectedLatch);
        } catch (IOException e) {
            LOGGER.error(e.getMessage(), e);
        }
    }

    private static void waitUntilConnected(ZooKeeper zk, CountDownLatch connectedLatch) {
        if (States.CONNECTING == zk.getState()) {
            try {
                connectedLatch.await();
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
        }
    }

    static class ConnectedWatcher implements Watcher {

        private CountDownLatch connectedLatch;

        ConnectedWatcher(CountDownLatch connectedLatch) {
            this.connectedLatch = connectedLatch;
        }

        @Override
        public void process(WatchedEvent event) {
            if (event.getState() == KeeperState.SyncConnected) {
                connectedLatch.countDown();
                String str = "Zookeeper service: " + zookeeperUrl + " 连接成功！";
                LOGGER.info(str);
                System.out.println(str);
            }
        }
    }

    public static ZookeeperUtils getInstance() {
        if (instance == null) {
            synchronized (ZookeeperUtils.class) {
                if (instance == null) {
                    instance = new ZookeeperUtils();
                }
            }
        }
        return instance;
    }

    public boolean create(String path, String data) {
        try {
            this.zk.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            return true;
        } catch (KeeperException e) {
            LOGGER.error(e.getMessage(), e);
        } catch (InterruptedException e) {
            LOGGER.error(e.getMessage(), e);
        }
        return false;
    }

    /**
     * 读取指定节点数据内容
     *
     * @param path 节点path
     * @return
     */
    public String read(String path) {
        try {
            return new String(this.zk.getData(path, false, null));
        } catch (KeeperException e) {
            LOGGER.error(e.getMessage(), e);
        } catch (InterruptedException e) {
            LOGGER.error(e.getMessage(), e);
        }
        return "";
    }

    /**
     * 更新指定节点数据内容
     *
     * @param path 节点path
     * @param data 数据内容
     * @return
     */
    public boolean write(String path, String data) {
        try {
            Stat stat = zk.setData(path, data.getBytes(), -1);
            LOGGER.info(stat.toString());
            return true;
        } catch (KeeperException e) {
            LOGGER.error(e.getMessage(), e);
        } catch (InterruptedException e) {
            LOGGER.error(e.getMessage(), e);
        }
        return false;
    }

    /**
     * 删除指定节点
     *
     * @param path 节点path
     */
    public void delete(String path) {
        try {
            this.zk.delete(path, -1);
        } catch (KeeperException e) {
            LOGGER.error(e.getMessage(), e);
        } catch (InterruptedException e) {
            LOGGER.error(e.getMessage(), e);
        }
    }

    public boolean exist(String path) {
        try {
            Stat stat = this.zk.exists(path, false);
            if (stat != null) {
                return true;
            }
        } catch (KeeperException e) {
            LOGGER.error(e.getMessage(), e);
        } catch (InterruptedException e) {
            LOGGER.error(e.getMessage(), e);
        }
        return false;
    }

    public void registerWatcher(final String znode, final ICallback callback) {
        try {
            this.zk.exists(znode, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    ZNodeEvent zNodeEvent = new ZNodeEvent();
                    zNodeEvent.setEventPath(event.getPath());
                    zNodeEvent.setEventType(event.getType().toString());
                    if (event.getType() == Event.EventType.None) {
                        switch (event.getState()) {
                            case SyncConnected:
                                break;
                            case Expired:
                                // It's all over
                                zNodeEvent.setStatus(ZNodeEvent.EventStatus.EXPIRED);
                                callback.call(zNodeEvent);
                                break;
                            default:
                                return;
                        }
                    }
                    String path = event.getPath();
                    if (path != null && path.equals(znode)) {
                        callback.call(zNodeEvent);
                    }
                }
            });
        } catch (KeeperException e) {
            LOGGER.error(e.getMessage(), e);
        } catch (InterruptedException e) {
            LOGGER.error(e.getMessage(), e);
        }

    }

    /**
     * 关闭ZK连接
     */
    public void close() {
        if (this.zk != null) {
            try {
                this.zk.close();
                instance = null;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
